﻿using System.Collections.Concurrent;
using System.Text;
using DotNetCore.CAP.Messages;
using Microsoft.AspNetCore.Http;
using Microsoft.Extensions.Logging;

namespace CapHttp.Core;

public class HttpConsumerMiddleware : IMiddleware
{
    private readonly ILogger<HttpConsumerMiddleware> _logger;
    
    private static readonly object Lock = new();

    private readonly Dictionary<string, List<string>> _topicGroups = new();
    private readonly ConcurrentDictionary<string, HttpConsumerClient> _consumerClients = new();

    public HttpConsumerMiddleware(ILogger<HttpConsumerMiddleware> logger)
    {
        _logger = logger;
    }

    internal void RegisterConsumerClient(HttpConsumerClient consumerClient, string groupName)
    {
        _consumerClients.TryAdd(groupName, consumerClient);
    }
    
    internal void Subscribe(string groupId, IEnumerable<string> topics)
    {
        lock (Lock)
        {
            foreach (var topic in topics)
            {
                if (_topicGroups.TryGetValue(topic, out var value))
                {
                    if (!value.Contains(groupId))
                    {
                        value.Add(groupId);
                    }
                }
                else
                {
                    _topicGroups.Add(topic, new List<string> { groupId });
                }    
            }
        }
    }
    
    internal void Unsubscribe(string groupId)
    {
        _consumerClients.Remove(groupId, out _);
        _logger.LogInformation("Removed consumer client! --> Group:"+ groupId);
    }
    
    public async Task InvokeAsync(HttpContext context, RequestDelegate next)
    {
        if (context.Request.Headers.TryGetValue(CapHttpConstants.API_NAME_HEADER, out var apiName))
        {
            context.Request.EnableBuffering();
            
            var reader = new StreamReader(context.Request.Body);
            var json = await reader.ReadToEndAsync();
            var buffer = Encoding.UTF8.GetBytes(json);
            context.Request.Body.Seek(0, SeekOrigin.Begin);

            Dictionary<string, string?> headers = new Dictionary<string, string?>();
            foreach (var (key, value) in context.Request.Headers)
            {
                headers.Add(key, value);
            }

            var name = context.Request.Headers[Headers.MessageName].ToString();
            
            lock (Lock)
            {
                if (_topicGroups.TryGetValue(name, out var groupList))
                {
                    foreach (var groupId in groupList)
                    {
                        if (_consumerClients.TryGetValue(groupId, out var consumerClient))
                        {
                            headers.TryAdd(Headers.Group, groupId);
                            var message = new TransportMessage(headers, buffer);
                            try
                            {
                                consumerClient.OnMessage(message);
                            }
                            catch (Exception e)
                            {
                                _logger.LogError(e, e.Message);
                                throw;
                            }
                        }
                    }
                }
                else
                {
                    throw new InvalidOperationException(
                        $"Cannot find the corresponding group for {name}. Have you subscribed?");
                }
            }
        }
        else
        {
            await next(context);
        }
    }
}